package cgl.granules.application;

import java.io.IOException;
import java.util.Properties;

import cgl.granules.dataset.DatasetCollection;
import cgl.granules.dataset.DatasetException;
import cgl.granules.exception.MarshallingException;
import cgl.granules.operation.MapReduceBase;
import cgl.granules.operation.MapReduceException;
import cgl.granules.results.ResultPayload;
import cgl.granules.results.Results;
import cgl.granules.dataset.StreamingAccess;
import cgl.granules.dataset.Dataset;

import cgl.narada.event.NBEvent;

/***
 * 
 * A map to hold POS data and to be binded with an RyoReduce instance.
 * 
 * @author Chao Sun
 * 
 */

public class PosMap extends MapReduceBase {

	private String datasetIdentifier = "POS-Data-Map";
	private Properties props = null;
	private DatasetCollection datasetCollection = null;
	private StreamingAccess access = null;
	private String synopsis = null;
	private ResultPayload resultPayload = null;

	public PosMap() {
		setDomain("Generic");
		setVersionInformation(200);
		setOperationIdentifier("Ryo-to-Pos-Map");
		resultPayload = ResultPayload.getInstance();
	}

	public void execute() {
		if (!hasDatasetCollection()) {
			System.out.println("No dataset initialized. exiting...");
			return;
		}

		props = getProcessingDirectives();
		if (!props.containsKey("DataType")
				|| !props.getProperty("DataType").equals("POS")) {
			System.out.println("Incorrect map! exiting...");
			return;
		}
		
		datasetCollection = getDatasetCollection();
		try {
			Dataset dataset = datasetCollection.getDataset(datasetIdentifier);

			if (dataset.getDatasetType() == Dataset.STREAMS) {
				access = (StreamingAccess) dataset;
			} else {
				System.out
						.println("Incorrect datatset: "
								+ dataset.getDatasetType()
								+ " initialized. exiting...");
				return;
			}

			if (access != null) {
				while (dataset.isDataAvailable()) {
					NBEvent nbEvent = access.getStreamingData();
					processStreamPacket(nbEvent);
					nbEvent = null;
				}
			} else {
				System.out.println("\nThe INPUT Dataset is NULL!\n");
			}

			datasetCollection = null;
			access = null;
			dataset = null;

		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	private void processStreamPacket(NBEvent nbEvent) throws IOException,
			MarshallingException, MapReduceException {
		synopsis = (String) nbEvent.getContentSynopsis();
		if (!synopsis.equals("POS-Data-Topic")) {
			System.out.println("Read data from a wrong topic");
			return;
		}

		byte[] payload = nbEvent.getContentPayload();
		String posData = "POS" + resultPayload.getString(payload);
		// System.out.println("\nPOS Map got data\n" + posData);
		byte[] resultPayloadBytes = resultPayload.getBytes(posData);

		Results results = createResults(true, true);
		results.setResultPayload(resultPayloadBytes);
		writeResults(results);

		payload = null;
		posData = null;
		resultPayloadBytes = null;
		results = null;
	}

}